package core.wc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Spark03_WordCount {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    val sc = new SparkContext(sparkConf)
    println("第一种方法")
//    wordCount1(sc)
    println("第二种方法")
//    wordCount2(sc)
    wordCount8(sc)

    sc.stop()
  }

  def wordCount1(sc:SparkContext):Unit={
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))
    val value: RDD[String] = rdd.flatMap(_.split(" "))
    val value1: RDD[(String, Int)] = value.map((_, 1))
    val value2: RDD[(String, Int)] = value1.reduceByKey(_+_)
    value2.collect().foreach(println)
  }

  def wordCount2(sc:SparkContext):Unit={
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))
    val value: RDD[String] = rdd.flatMap(_.split(" "))
    val value1: RDD[(String, Iterable[String])] = value.groupBy(word => word)
    val value2: RDD[(String, Int)] = value1.mapValues(iter => iter.size)
    value2.collect().foreach(println)
  }

  def wordCount3(sc: SparkContext): Unit = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))
    val value: RDD[String] = rdd.flatMap(_.split(" "))
    val value1: RDD[(String, Int)] = value.map((_, 1))
    val value2: RDD[(String, Int)] = value1.aggregateByKey(0)(_ + _,_+_)
    value2.collect().foreach(println)
  }

  def wordCount4(sc: SparkContext): Unit = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))
    val value: RDD[String] = rdd.flatMap(_.split(" "))
    val value1: RDD[(String, Int)] = value.map((_, 1))
    val value2: RDD[(String, Int)] = value1.foldByKey(0)(_ + _)
    value2.collect().foreach(println)
  }

  def wordCount5(sc: SparkContext): Unit = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))
    val value: RDD[String] = rdd.flatMap(_.split(" "))
    val value1: RDD[(String, Int)] = value.map((_, 1))
    val value2: RDD[(String, Int)] = value1.combineByKey(
      v=>v,
      _ + _,
      _ + _
    )
    value2.collect().foreach(println)
  }

  def wordCount6(sc: SparkContext): Unit = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))
    val value: RDD[String] = rdd.flatMap(_.split(" "))
    val value1: RDD[(String, Int)] = value.map((_, 1))
    val value2: collection.Map[String, Long] = value1.countByKey()
    value2.foreach(println)
  }

  def wordCount7(sc: SparkContext): Unit = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))
    val value: RDD[String] = rdd.flatMap(_.split(" "))
    val value2: collection.Map[String, Long] = value.countByValue()
    value2.foreach(println)
  }

  def wordCount8(sc: SparkContext): Unit = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))
    val value: RDD[String] = rdd.flatMap(_.split(" "))
    val mapWord: RDD[mutable.Map[String, Long]] = value.map(
      word => {
        mutable.Map[String, Long]((word, 1L))
      }
    )
    val stringToLong: mutable.Map[String, Long] = mapWord.reduce(
      (map1, map2) => {
       map2.foreach {
          case (word, count) => {
            val newCount = map1.getOrElse(word, 0L) + count
            map1.update(word, newCount)
          }
        }
        map1
      }
    )
    println(stringToLong)
  }
}
